package com.xrui.hbase.schema;

import com.google.common.base.Preconditions;
import com.moz.kiji.delegation.Lookups;

import com.xrui.hbase.HBase;
import com.xrui.hbase.exception.InvalidLayoutException;
import com.xrui.hbase.exception.TableLayoutNotFoundException;
import com.xrui.hbase.util.RegexHashMap;
import com.xrui.hbase.util.Resources;
import org.apache.hadoop.conf.Configuration;
import org.codehaus.jackson.JsonFactory;
import org.codehaus.jackson.JsonNode;
import org.codehaus.jackson.JsonParser;
import org.codehaus.jackson.map.ObjectMapper;
import org.codehaus.jackson.type.TypeReference;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.io.IOException;
import java.io.Reader;
import java.util.HashMap;
import java.util.Map;

public final class TableLayoutsParser {
    /**
     * The default layout resource file name
     */
    public static String DEFAULT_LAYOUT_RESOURCE = "hbase_table_schema.json";
    private final static Logger LOG = LoggerFactory.getLogger(TableLayoutsParser.class);
    /**
     * A map from table names to their layouts.
     */
    private final Map<String, TableLayoutResolver> mResolverCache;

    private RegexHashMap<JsonNode> mRegexTableSpecs;

    /**
     * Constructs an TableLayoutsParser for table actions.
     *
     * @param parser The json parser to parse the table layouts.
     * @throws InvalidLayoutException If the layout specification json is invalid.
     */
    private TableLayoutsParser(JsonParser parser) throws InvalidLayoutException {
        mResolverCache = new HashMap<>();
        final ObjectMapper mapper = new ObjectMapper();
        try {
            JsonNode rootNode = mapper.readTree(parser);
            JsonNode tablesNode = rootNode.findValue("tables");
            mRegexTableSpecs = mapper.readValue(tablesNode, new TypeReference<RegexHashMap<JsonNode>>() {
            });
        } catch (IOException e) {
            throw new InvalidLayoutException(e.getMessage());
        }
    }

    /**
     * Constructs an TableLayoutsParser.
     *
     * @return an {@code TableLayoutsParser} for the provided json.
     * @throws InvalidLayoutException If the layout specification json is invalid.
     */
    public static TableLayoutsParser newParser() throws InvalidLayoutException {
        return newParser(Resources.openSystemTextResource(DEFAULT_LAYOUT_RESOURCE));
    }

    /**
     * Constructs an TableLayoutsParser.
     *
     * @param json The json whose table layouts are to be parsed.
     * @return an {@code TableLayoutsParser} for the provided json.
     * @throws InvalidLayoutException If the layout specification json is invalid.
     */
    public static TableLayoutsParser newParser(String json) throws InvalidLayoutException {
        JsonParser parser;
        try {
            parser = new JsonFactory().createJsonParser(json)
                .enable(JsonParser.Feature.ALLOW_COMMENTS)
                .enable(JsonParser.Feature.ALLOW_SINGLE_QUOTES)
                .enable(JsonParser.Feature.ALLOW_UNQUOTED_FIELD_NAMES);
        } catch (IOException e) {
            throw new InvalidLayoutException(e.getMessage());
        }
        return new TableLayoutsParser(parser);
    }

    /**
     * Constructs an TableLayoutsParser.
     *
     * @param reader The reader to parse the table layouts.
     * @return an {@code TableLayoutsParser} for the provided json.
     * @throws InvalidLayoutException If the layout specification json is invalid.
     */
    public static TableLayoutsParser newParser(Reader reader) throws InvalidLayoutException {
        JsonParser parser;
        try {
            parser = new JsonFactory().createJsonParser(reader)
                .enable(JsonParser.Feature.ALLOW_COMMENTS)
                .enable(JsonParser.Feature.ALLOW_SINGLE_QUOTES)
                .enable(JsonParser.Feature.ALLOW_UNQUOTED_FIELD_NAMES);
        } catch (IOException e) {
            throw new InvalidLayoutException(e.getMessage());
        }
        return new TableLayoutsParser(parser);
    }

    public TableLayout get(String tableName, Configuration configuration,HBase hbase) throws TableLayoutNotFoundException {
        JsonNode tableSpec = mRegexTableSpecs.get(tableName);

        try {
            Preconditions.checkNotNull(tableSpec);
            String resolverName = tableSpec.get("resolver").getTextValue();
            if (!mResolverCache.containsKey(resolverName)) {
                TableLayoutResolver resolver = Lookups.getNamed(TableLayoutResolver.class).lookup(resolverName);
                mResolverCache.put(resolverName, resolver);
            }
            TableLayoutResolver resolver = mResolverCache.get(resolverName);
            if (tableSpec.get("isStatic").asBoolean()) {
                return resolver.resolve(tableName, tableSpec.get("spec"));
            } else {
                return resolver.resolve(tableName, configuration);
            }
        } catch (Exception e) {
            LOG.error("Exception when resolving table layout", e);
            throw new TableLayoutNotFoundException(tableName);
        }
    }
}
